package timeandwindow;

/*
* Flink对迟到数据的处理
*   todo 1.延迟水位线的推进, 生成水位线的时候人为的减去一个延迟时间(经过监控，确定绝大多数的数据的延迟时间 ， 保证绝大多数的数据能正确参与窗口计算)
*
*   todo 2.延迟窗口的关闭,窗口触发计算输出结果， 窗口不关闭，再等待一段时间（保证少量的迟到更久的数据能正确的参与窗口计算）
*
*   todo 3.使用侧输出流接收极端数据（保证个别的迟到特别久的数据不丢失，再想办法对该部分数据进行额外的处理）
*
* */

import com.atguigu.pojo.Event;
import com.atguigu.pojo.UrlViewCount;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

public class Flink11_LateData {
     public static void main(String[] args) {
             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
             env.setParallelism(1);

             env.getConfig().setAutoWatermarkInterval(500L);

        SingleOutputStreamOperator<Event> ds = env.socketTextStream("hadoop102", 8888)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new Event(fields[0].trim(), fields[1].trim(), Long.valueOf(fields[2].trim()));
                        }

                ).assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2)) // 1. 延迟水位线的推进 ，
                                .withTimestampAssigner(//todo 这是一个时间戳分配生成器
                                        (event, ts) -> event.getTs()
                                )
                );

        ds.print("INPUT");

         OutputTag<Event> lateOutputTag = new OutputTag<>("lateOutputTag", Types.POJO(Event.class));

         //需求：统计每10秒内每个url的点击数
         //窗口：键值分区事件时间滚动窗口

         ds
                 .keyBy( Event -> Event.getUrl())
                 .window(
                         TumblingEventTimeWindows.of(Time.seconds(10)) //10秒一个窗口
                 )
                 .allowedLateness(Time.seconds(5)) //todo 2.延迟窗口的关闭
                 .sideOutputLateData(lateOutputTag)//todo 3.使用侧输出流接收极端数据
                 .aggregate(
                         new AggregateFunction<Event, UrlViewCount, UrlViewCount>() {
                             @Override
                             public UrlViewCount createAccumulator() {
                                 return new UrlViewCount(null,null,null,0L);
                             }

                             @Override
                             public UrlViewCount add(Event value, UrlViewCount accumulator) {

                                 accumulator.setCount(accumulator.getCount() + 1);
                                 return accumulator;
                             }

                             @Override
                             public UrlViewCount getResult(UrlViewCount accumulator) {
                                 return accumulator;
                             }

                             @Override
                             public UrlViewCount merge(UrlViewCount urlViewCount, UrlViewCount acc1) {return null;}

                         },
                             new ProcessWindowFunction<UrlViewCount,UrlViewCount,String, TimeWindow>(){


                                 @Override
                                 public void process(String key, ProcessWindowFunction<UrlViewCount, UrlViewCount, String, TimeWindow>.Context context, Iterable<UrlViewCount> elements, Collector<UrlViewCount> out) throws Exception {
                                      //获得增量聚合的结果
                                     UrlViewCount urlViewCount = elements.iterator().next();
                                     //设置url
                                     urlViewCount.setUrl(key);
                                     //设置窗口信息
                                     urlViewCount.setWindowStart(context.window().getStart());
                                     urlViewCount.setWindowEnd(context.window().getEnd());
                                     //输出
                                     out.collect(urlViewCount);
                                 }
                             }
                 )
                 .print();







             try {
                 env.execute();
             } catch (Exception e) {
                 throw new RuntimeException(e);
             }
         }
}
